package com.game.framework;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.pattern.Patterns;
import akka.util.Timeout;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;
import scala.concurrent.duration.FiniteDuration;

import java.util.Arrays;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/// Actor管理器
@Setter
@Slf4j
public class ActorManager implements IManager {
    @Getter
    private static ActorManager instance = new ActorManager();
    @Getter
    private final ActorSystem actorSystem;

    public ActorManager() {
        actorSystem = ActorSystem.create("game-actor-system");
    }

    private static final ConcurrentHashMap<Integer, ActorRef> actorRefs = new ConcurrentHashMap<>();
    protected static Integer getKey(Class<? > aClass, Object... args) {
        return Objects.hash(aClass.getSimpleName(), Arrays.hashCode(args));
    }

    /// 注册
    protected static void register(Class<? > aClass, Object... args) {
        Props props = Props.create(aClass, args);
        Integer key = getKey(aClass, args);
        ActorRef actorRef = getInstance().getActorSystem().actorOf(props, key.toString());
        actorRefs.put(key, actorRef);
    }

    /// 获取Actor
    public static ActorRef getActor(Class<? > aClass, Object... args) {
        Integer key = getKey(aClass, args);

        ActorRef actorRef = actorRefs.get(key);
        if (actorRef == null) {
            synchronized (key.toString()) {
                if (actorRefs.get(key) == null) {
                    register(aClass, args);
                }
                actorRef = actorRefs.get(key);
                return actorRef;
            }
        }
        return actorRef;
    }

    public static <T> T ask(Class<? > aClass, Object msg, Object... args) {
        log.debug("Actor ask, class: {}, msg: {}", aClass.getSimpleName(), msg);
        ActorRef actor = getActor(aClass, args);
        FiniteDuration timeout = scala.concurrent.duration.Duration.create(10, TimeUnit.SECONDS);
        Future<Object> future = Patterns.ask(actor, msg, Timeout.apply(timeout));
        try {
            //noinspection unchecked
            return (T) Await.result(future, Duration.create("10 seconds"));
        } catch (InterruptedException | java.util.concurrent.TimeoutException e) {
            log.error("Actor ask timeout or interrupted, class: {}, msg: {}", aClass.getSimpleName(), msg, e);
            throw new RuntimeException("Actor communication failed", e);
        }
    }



    @Override
    public void shutdown() {

    }

    @Override
    public void init() {

    }

    /// 10分钟显示一次actor 数量
    public static void showActorCount() {
        log.info("Actor count: {}", actorRefs.size());
        for (Integer key : actorRefs.keySet()) {
            log.info("Actor key: {}, value: {}", key, actorRefs.get(key));
        }
    }
}
